package com.pool.thread;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author linyu.dai
 * @project rpc
 * @description
 * @creat 2022/12/1 20:24:58
 */
@Slf4j
public class ThreadPoolTrader implements Executor {

    private final AtomicInteger ctl = new AtomicInteger(0);

    //核心线程数
    private volatile int corePoolSize;

    //最大线程数
    private volatile int maximumPoolSize;

    //阻塞队列
    private final BlockingQueue<Runnable> workQueue;

    private final AtomicBoolean isDown = new AtomicBoolean(false);


    public ThreadPoolTrader(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
    }

    /**
     * 回调任务
     */
    public <T> Future<T> execute(Callable<T> task){
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = new FutureTask<>(task);
        execute(ftask);
        return ftask;
    }

    /**
     * 添加任务
     */
    @Override
    public void execute(Runnable command) {
        int c = ctl.get();
        if (c < corePoolSize) {
            if (!addWorker(command)) {
                reject();
            }
            return;
        }
        if (!workQueue.offer(command)) {
            if (!addWorker(command)) {
                reject();
            }
        }
    }

    private boolean addWorker(Runnable firstTask) {
        if (ctl.get() >= maximumPoolSize) return false;
        Worker worker = new Worker(firstTask);
        worker.thread.start();
        ctl.incrementAndGet();
        return true;
    }

    public void shutdown(){
        log.info("关闭连接池");
        log.info("当前线程个数:{}", ctl.get());
        isDown.getAndSet(true);
        //推送空任务通知关闭
        for (int i = 0; i < ctl.get(); i++) {
            workQueue.offer(() -> {});
        }
    }


    private final class Worker implements Runnable {

        final Thread thread;
        Runnable firstTask;

        public Worker(Runnable firstTask) {
            this.thread = new Thread(this);
            this.firstTask = firstTask;
        }

        @Override
        public void run() {
            Runnable task = firstTask;
            try {
                while (task != null || (task = getTask()) != null) {
                    task.run();
                    //如果线程数大于 核心线程数 执行完就释放
                    if (ctl.get() > corePoolSize || isDown.get()) {
                        break;
                    }
                    task = null;
                }
            } finally {
                ctl.decrementAndGet();
            }
        }

        private Runnable getTask() {
            for (; ; ) {
                try {
                    System.out.println("workQueue.size：" + workQueue.size());
                    return workQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void reject() {
        log.error("Error！ctl.count：" + ctl.get() + " workQueue.size：" + workQueue.size());
    }
}
